package main.prodcons;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * ProducerConsumer
 * <p/>
 * Producer and consumer tasks in a desktop search application
 * 
 * @author Brian Goetz and Tim Peierls
 */
public class ProducerConsumer2 {

	public static void main(String[] args) throws InterruptedException {
		File[] roots = { 
				new File("C:\\db.0002"),
				new File("C:\\DRIVERS"),
				new File("C:\\eclipse"),
				new File("C:\\Intel"),
				new File("C:\\Program Files"),
				new File("C:\\Program Files (x86)"),
				new File("C:\\"),
		};
		startIndexing(roots);
	}

	static class FileCrawler implements Runnable {
		private final List<LinkedBlockingDeque<File>> queuesList;
		private final File root;

		private final ExecutorService executor;

		private final AtomicInteger latch;

		public FileCrawler(List<LinkedBlockingDeque<File>> queuesList,
				File root, ExecutorService executor,
				final AtomicInteger crawlerLatch) {
			this.queuesList = queuesList;
			this.root = root;
			this.executor = executor;
			this.latch = crawlerLatch;
		}

		private boolean alreadyIndexed(File f) {
			return false;
		}

		@Override
		public void run() {
			try {
				crawl(root);
			} catch (InterruptedException e) {
				e.printStackTrace();
				Thread.currentThread().interrupt();
			} finally {
				latch.decrementAndGet();
			}
		}

		private void crawl(File root) throws InterruptedException {
			for (File entry : root.listFiles())
				if (entry.isDirectory()) {
					latch.incrementAndGet();
					executor.submit(new FileCrawler(queuesList, entry,
							executor, latch));
				} else if (!alreadyIndexed(entry)) {
					queuesList.get(getQueueIndex(entry)).put(entry);
				}
		}

		private int getQueueIndex(File file) {
			return Math.abs(file.hashCode()) % (queuesList.size());
		}
	}

	static class Indexer implements Runnable {
		private final List<LinkedBlockingDeque<File>> queuesList;
		private final int index;
		private final CountDownLatch latch;

		private int totalCount;
		private int selfCount;
		private int friendCount;

		final ExecutorService executor;

		private volatile boolean isInterrupted = false;

		public Indexer(List<LinkedBlockingDeque<File>> queuesList, int index,
				ExecutorService executor, CountDownLatch latch) {
			this.queuesList = queuesList;
			this.index = index;
			this.latch = latch;
			this.executor = executor;
		}

		@Override
		public void run() {
			BlockingDeque<File> myQueue = queuesList.get(index);

			while (!myQueue.isEmpty()) {
				File file = myQueue.poll();
				if (file != null) {
					indexFile(file);
					selfCount++;
				}
			}

			if (isInterrupted) {
				System.out.println("Interrupted! : " + index + " : "
						+ totalCount + " : " + selfCount + " : " + friendCount);
				latch.countDown();
				return;
			}

			for (BlockingDeque<File> queue : queuesList) {
				if (queue == myQueue)
					continue;
				if (processQueue(queue)) {
					break;
				}
			}

			executor.submit(this);

		}

		private boolean processQueue(BlockingDeque<File> queue) {
			File file = queue.pollLast();
			if (file != null) {
				indexFile(file);
				friendCount++;
			}
			return file == null;
		}

		private void indexFile(File file) {
			// Index the file...
			System.out.println(index + " : " + file.getAbsolutePath());
			totalCount++;
		};

		public void interrupt() {
			isInterrupted = true;
		}
	}

	private static final int BOUND = 100;
	private static final int N_CONSUMERS = 4;// Runtime.getRuntime().availableProcessors();

	public static void startIndexing(File[] roots) throws InterruptedException {

		ArrayList<LinkedBlockingDeque<File>> queuesList = new ArrayList<LinkedBlockingDeque<File>>(
				N_CONSUMERS);

		ArrayList<Indexer> indexeres = new ArrayList<>();

		for (int i = 0; i < N_CONSUMERS; ++i) {
			queuesList.add(new LinkedBlockingDeque<File>(BOUND));
		}

		CountDownLatch indexersLatch = new CountDownLatch(N_CONSUMERS);

		ExecutorService executor = Executors.newFixedThreadPool(31);
		ExecutorService indexersExecutor = Executors.newFixedThreadPool(4);

		AtomicInteger crawlerLatch = new AtomicInteger(0);

		long time = System.currentTimeMillis();

		for (File root : roots) {
			crawlerLatch.incrementAndGet();
			executor.submit(new FileCrawler(queuesList, root, executor,
					crawlerLatch));
		}

		for (int i = 0; i < N_CONSUMERS; i++) {
			Indexer indexer = new Indexer(queuesList, i, indexersExecutor,
					indexersLatch);
			indexeres.add(indexer);
			indexersExecutor.submit(indexer);
		}

		while (crawlerLatch.get() != 0) {

		}

		System.err.println("crawlers ended");

		for (Indexer indexer : indexeres) {
			indexer.interrupt();
		}

		indexersLatch.await();

		System.out.println("Time: " + (System.currentTimeMillis() - time));

		executor.shutdown();
		indexersExecutor.shutdown();
	}

}